package com.navinfo.tripanalysis.service.impl;

import com.navinfo.tripanalysis.pojo.LoadDataParam;
import com.navinfo.tripanalysis.service.LoadAbnormalTripStatictisDataService;
import lombok.Data;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Data
public class LoadAbnormalTripStatictisServiceImpl implements LoadAbnormalTripStatictisDataService {
    private static final Logger logger = LoggerFactory.getLogger(LoadAbnormalProtocolPointServiceImpl.class);
    private String S_DBname;
    private String S_Tablename;
    private String T_DBname;
    private String T_Tablename;
    private String Temptable;

    @Override
    public void insert(SparkSession spark, JavaSparkContext jsc, LoadDataParam param) {

        StringBuilder sb=new StringBuilder();
        sb.append("insert into table ");
        if(StringUtils.isNotEmpty(T_DBname)){
            sb.append(T_DBname).append(".");
        }
        sb.append(T_Tablename);
        sb.append("\n");
        sb.append("select trip.tid,trip.tid,trip.route_start_time,trip.route_end_time,'6',");
        sb.append("trip.trip_mileage/1000,trip.trip_fuel/1000,fuel_100km,");
        sb.append("'-1','-1','-1','-1','-1','-1','-1','-1','-1','default'");
        sb.append("\n");
        sb.append(" from ");
        if(StringUtils.isNotEmpty(S_DBname)){
            sb.append(S_DBname).append(".");
        }
        sb.append(S_Tablename);
        sb.append(" as trip");
        sb.append("\n");
        sb.append("inner join ");
        sb.append(Temptable);
        sb.append(" as temp");
        sb.append("\n");
        sb.append("on trip.tid = temp.tid and trip.tripid = temp.tripid");
        sb.append(" where trip.day =  ");
        sb.append(param.getDay());

        String sql = sb.toString();
        logger.error("load abnormaltrip sql:-----------------\n{}",sql);

        spark.sql(sql);

    }
}
